package com.udf.flink.scala.demo

import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}

object SQLRemote {
  def main(args: Array[String]): Unit = {
    // 创建一个批处理的执行环境
//    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createRemoteEnvironment("192.168.58.102",8081)
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance().inStreamingMode().build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, settings)
    val sql =
      """CREATE TABLE Orders (
                b INT,
                c INT,
                a STRING,
                rowtime AS localtimestamp,
                WATERMARK FOR rowtime AS rowtime
               ) WITH (
                'connector' = 'datagen',
                'rows-per-second'='10',
                'fields.b.kind'='sequence',
                'fields.b.start'='1',
                'fields.b.end'='20000',
                'fields.c.min'='1',
                'fields.c.max'='100',
                'fields.a.length'='2'
               )"""


//    val rs_sql =
//      """
//             CREATE TABLE rs (
//                 a STRING,
//                  b INT,
//                  c INT,
//                  ts  TIMESTAMP(3)
//             ) WITH (
//                 'connector' = 'print'
//             )
//         """
//    tableEnv.executeSql(sql);
//    tableEnv.executeSql(rs_sql)
//    tableEnv.executeSql("insert into rs(a,b,c,ts) select a,b,c,rowtime from Orders") //.await()

    val rs_sql2 =
      """
             CREATE TABLE rs2 (
                 a STRING,
                  b INT
             ) WITH (
                 'connector' = 'print'
             )
         """
    tableEnv.executeSql(sql);
    tableEnv.executeSql(rs_sql2)
    tableEnv.executeSql("insert into rs2(a,b) select a,sum(b) b from Orders group by a") //.await()
  }
}
